package org.dingfugui.triple.protocol.binary.aio;

import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.util.LinkedList;
import java.util.Queue;

/**
 * Created by fugui.ding on 2015/4/30.
 */
public class AioDataWriter {

    private AsynchronousSocketChannel socket;

    private Queue<ByteBuffer> queue = new LinkedList<>();

    private boolean isWritting;

    public AioDataWriter(AioProducerConnection con) {
        this.socket = con.socket;
    }

    public void writeData(byte[] data) {
        writeData(ByteBuffer.wrap(data));
    }

    private void writeData(ByteBuffer bb) {

        boolean threadShouldWrite = false;

        synchronized (queue) {
            queue.add(bb);
            if (!isWritting) {
                isWritting = true;
                threadShouldWrite = true;
            }
        }

        if (threadShouldWrite) {
            writeFromQueue();
        }

    }

    private void writeFromQueue() {
        ByteBuffer bb;

        synchronized (queue) {
            bb = queue.poll();
            if (bb == null) {
                isWritting = false;
            }
        }
        if (isWritting) {
            writeBuffer(bb);
        }
    }

    private void writeBuffer(ByteBuffer bb) {
        socket.write(bb, bb, new CompletionHandler<Integer, ByteBuffer>() {
            @Override
            public void completed(Integer result, ByteBuffer bb) {
                if (bb.hasRemaining()) {
                    socket.write(bb, bb, this);
                } else {
                    writeFromQueue();
                }
            }

            @Override
            public void failed(Throwable exc, ByteBuffer attachment) {

            }
        });
    }
}
